package com.huwaiwai.kafka.demo01.producer;

import com.huwaiwai.kafka.KafkaProperties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

/**
 * 生产者
 * Created by Administrator on 2017/4/29.
 */
public class KafkaProducer extends Thread{

    private Properties props = new Properties();
    private Producer<Integer, String> producer;
    private String topic;
    private final int SLEEP = 1000 * 3;

    public KafkaProducer(String topic) {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", KafkaProperties.BROKER_LIST);
        producer = new Producer<>(new ProducerConfig(props));
        this.topic = topic;
    }

    @Override
    public void run() {
        int offsetNo = 1;
        while (true) {
            String msg = new String("Message_" + offsetNo);
            System.out.println("Send->[" + msg + "]");
            producer.send(new KeyedMessage<>(topic, msg));
            offsetNo++;
            try {
                sleep(SLEEP);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
